package cgl.narada.transport.rtp;
/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or GlobalMMCS,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called GlobalMMCS,
 * nor may Indiana University or Community Grids Lab or GlobalMMCS appear
 * in their name, without prior written permission of ARTI.
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.ArrayList;

import cgl.narada.matching.rtp.RtpEvent;
import cgl.narada.matching.rtp.RtpProfile;
import cgl.narada.protocol.Destinations;
import cgl.narada.protocol.Protocol;
import cgl.narada.transport.Link;
import cgl.narada.transport.LinkPerformanceData;
import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.TransportHandler;

/** An RTP implementation of the Link interface. To deal with raw RTP clients
    we need to establish two underlying communication paths. One is for the
    rtp data packets and the other is for rtcp control packets.

    This RTPLink class is used for unicast UDP communications.
    This link can subscribe to one or more topics.
    It publishes the received data on one topic pair.
    It publishes the rtp data on the given even numbered topic and rtcp data on the following odd numbered topic.

    @author Ahmet Uyar
    $Revision $
    $Date$
*/

public class RTPLink implements Link, TransportDebugFlags {

   private String linkId;
   private Object naradaBrokeringId;
   private String linkType = "rtp";
   private boolean gatherPerformance = false;
   private boolean secure = false;
   private String securityInformation = "Communication on this link is not encrypted";
   private boolean connectionLost = false;
   private boolean linkClosed = false;
   private String moduleName = "RTPLink: ";

   private RTPSender dataSender;
   private RTPSender controlSender;
   private RTPReceiver dataReceiver;
   private RTPReceiver controlReceiver;

   private RTPLinkFactory rtpLinkFactory;
   private TransportHandler transportHandler;

   private int nodeIdentifier = 0;
   private String brokerLocation;
   private Object synchObject = new Object();
   private int ttl = 64; // default ttl value

   private ArrayList subscriptionList; // this will hold a list of Integer numbers
                                      // it shows the topics this RTPLink subscribes

   /**
    * create an RTPLink with the given parameters.
    * InetAddress must be a unicast address.
    */
   public RTPLink(InetAddress remoteAddress,
                  int remotePort,
                  int localPort,
                  int publicationTopic,
                  RTPLinkFactory rtpLinkFactory)
         throws Exception {

      this.transportHandler = rtpLinkFactory.getTransportHandler();
      this.rtpLinkFactory = rtpLinkFactory;

      DatagramSocket dataSocket = null;
      DatagramSocket controlSocket = null;;

      try{
         if(remoteAddress.isMulticastAddress()){
            MulticastSocket multicastDataSocket = new MulticastSocket(remotePort);
            multicastDataSocket.joinGroup(remoteAddress);
            multicastDataSocket.setTimeToLive( ttl );
            dataSocket = multicastDataSocket;

            MulticastSocket multicastControlSocket = new MulticastSocket(remotePort+1);
            multicastControlSocket.joinGroup(remoteAddress);
            multicastControlSocket.setTimeToLive( ttl );
            controlSocket = multicastControlSocket;
         }else{
            dataSocket = new DatagramSocket(localPort);
            controlSocket = new DatagramSocket(localPort+1);
         }
      }catch(Exception se){
         se.printStackTrace();
         throw new Exception(se.getMessage());
      }

      // set the linkId for this link
      linkId = RTPLinkFactory.createLinkID(remoteAddress, localPort);

      dataSender = new RTPSender(dataSocket, remoteAddress, remotePort);
      controlSender = new RTPSender(controlSocket, remoteAddress, remotePort + 1);

      dataReceiver = new RTPReceiver(linkId, dataSocket, publicationTopic, transportHandler);
      String controlLinkId = RTPLinkFactory.createLinkID(remoteAddress, localPort+1);
      controlReceiver = new RTPReceiver(controlLinkId, controlSocket, publicationTopic+1, transportHandler);

      dataReceiver.start();
      controlReceiver.start();

      if (RTPLink_Debug) {
         System.out.println(moduleName + " created [" + remoteAddress + ":" + remotePort);
      }

      subscriptionList = new ArrayList();
   }

   /** The method used to send data over a communication link */
   public void sendData(byte[] data) throws TransportException {
      if (RTPLink_Debug) {
//         System.out.println(moduleName + "Sending data");
      }

      if (nodeIdentifier == 0) {
         if (data[0] == cgl.narada.protocol.Protocol.NAP_CLIENT_ADD_RESPONSE) {
            processNodeAddressResponse(data);

            if (nodeIdentifier != 0) {

               synchronized(synchObject){
                  for (int i = 0; i < this.subscriptionList.size(); i++) {
                     int topic = ((Integer)subscriptionList.get(i)).intValue();
                     transportHandler.dataReceived( createSubscription(topic), linkId );
                     transportHandler.dataReceived( createSubscription(topic + 1), linkId );
                  }

               }
            }
         }

         return;
      }

      // create an RTPEvent and marshal it to send accross
      RtpEvent rtpEvent = new RtpEvent(data);
      byte[] dataToSend = rtpEvent.getData();

      if (dataToSend == null) {
         System.out.println(moduleName + "The unmarshalled data To send is NULL");
         return;
      }

      if ( rtpEvent.getTopic() % 2 == 0 ) {
         dataSender.sendData(dataToSend);
      } else {
         controlSender.sendData(dataToSend);
//         System.out.println("control package is sent to legacy rtp client .... .... ... ");
      }
   }

   /** The method used to send data over a communication link, this method
       also includes information regarding the type (audio/video/xml-attachments
       etc.) of the data encapsulated within the data stream. Using this
       information links can optimally flush/send data across the link
   */
   public void sendData(byte[] data, String dataType) throws TransportException {
      dataSender.sendData(data, dataType);
   }

   /** Sets the Id of the communication link */
   public void setLinkId(String stringId) {
      this.linkId = stringId;
   }

   /** Retrieves the String based id of the link */
   public String getLinkId() {
      return linkId;
   }

   /** The Naradabrokering Id of the node this link interacts with */
   public void setNaradaBrokeringId(Object brokeringId) {
      naradaBrokeringId = brokeringId;
   }

   /** Returns the NaradaBrokeringId associated with this link */
   public Object getNaradaBrokeringId() {
      return naradaBrokeringId;
   }

   /** Returns the type of the communication link */
   public String getLinkType() {
      return linkType;
   }

   /** Sets the conditions under which communication protocols need to be
       switched and the transport link that it should be switched to under
       the condition. Multiple constraints can be set signifying migration to
       different protocols.  */
   public void setLinkMigrationConstraint(String constraint, String linkType) {
      /**NBNI*/
   }

   /** Check the status of the communication link at intervals specified in
       the argument of this method */
   public void setLinkStatusInterval(long interval) {
      /**NBNI*/
   }

   /** Used to enable or disable performance gathering on a communication
       link */
   public void setPerformanceGathering(boolean gather) {
      gatherPerformance = gather;
   }

   /** Used to test weather performing gathering is enabled for a given
       communication link */
   public boolean performanceGatheringEnabled() {
      return gatherPerformance;
   }

   /** Returns the list of performance factors that are measured on this
       communication link */
   public String[] getPerformanceFactorsMeasured() {
      /**NBNI*/
      return null;
   }

   /** Returns the complete set of performance factors measured and their
       corresponding values*/
   public LinkPerformanceData[] getAllPerformanceData() {
      /**NBNI*/
      return null;
   }

   /** Returns the value of a performance factor as a string */
   public String getPerformanceData(String factorName) {
      /**NBNI*/
      return null;
   }

   /** Set the amount of time after which resources associated with this
       link need to be garbage collected. A value of '0' indicates that
       this link needs to be active at all times. */
   public void setKeepAliveTime(long keepAliveTime) {
      /**NBNI*/
   }

   /** Garbage collect resources associated with this link. Sockets need to
       be closed and any resources associated with communication, needs to be
       garbage collected */
   public void closeLink() {

      linkClosed = true;

      // close receiver threads
      dataReceiver.close();
      controlReceiver.close();

      // close sender threads
      dataSender.close();
      controlSender.close();
   }

   /** Indicates weather communication is encrypted on this link */
   public boolean isSecure() {
      return secure;
   }

   /** Information pertaining to the security strategy adopted for this
       communication link */
   public String getSecurityInformation() {
      return securityInformation;
   }

   /** If the senderNotification is false, it implies that the detection of
       connection loss has been reported by the receiver thread. Proceed to
       garbage collect the sender thread.
       If the boolean variable is true, it implies that the receiver thread
       has to be garbage collected. */
   public void processLossOfSocketConnection(boolean senderNotification) {
      if (linkClosed) {
         /** No need to do further processing */
         return;
      }

      if (connectionLost) {
         /** The loss was connection has been detected earlier, and the
         associated garbage collection operations are being performed */
         return;
      }

      connectionLost = true;
      rtpLinkFactory.reportLinkLoss(this);

      System.out.println(moduleName + "Processing of communication loss on link [" + linkId + " complete");
   }

   public TransportHandler getTransportHandler() {
      return transportHandler;
   }

   private void processNodeAddressResponse(byte[] addressResponse) {
      if (addressResponse.length == 1) {
         System.out.println("The broker node hasn't been assigned an address.. ");
      }

      if (addressResponse.length == 2) {
         System.out.println("The system has exceeded its capacity to handle" + " clients, please try at a different node ");
      }

      int offset = 0;
      offset++;

      byte[] destinationBytes = new byte[4];
      System.arraycopy(addressResponse, offset, destinationBytes, 0, 4);
      nodeIdentifier = cgl.narada.util.ByteUtilities.getInt(destinationBytes);

      byte[] serverDestBytes = new byte[addressResponse.length - offset];
      System.arraycopy(addressResponse, offset, serverDestBytes, 0, serverDestBytes.length);

      Destinations brokerDestination = new Destinations(serverDestBytes);
      brokerLocation = brokerDestination.toString();
      System.out.println("This RTPLink has been assigned the address " + nodeIdentifier);
   }

   /**
    * the topicNumber to subscribe.
    * this method can be invoked more than once and an RTPLink can subscribe to many topics
    */
   public String subscribe(int topicNumber){

      // if topicNumber is not an even number, then return an error String
      if(topicNumber%2 != 0)
         return "The given topic number is not even: "+topicNumber;

      // if a node identifier is not assigned yet to this link, add the topicNumber to
      // subscriptionList and return
      if(nodeIdentifier == 0){
         synchronized(synchObject){
            subscriptionList.add(new Integer(topicNumber));
         }
         return null;
      }

      // subscribe to this topicNumber
      transportHandler.dataReceived( createSubscription(topicNumber), linkId );
      transportHandler.dataReceived( createSubscription(topicNumber + 1), linkId );
      subscriptionList.add(new Integer(topicNumber));
      return null;
   }

   /**
    * the topicNumber to unsubscribe.
    */
   public String unsubscribe(int topicNumber){

      // if subscriptionList does not have this topic return a String
      if( !subscriptionList.contains(new Integer(topicNumber)) )
         return "This topic is not being subscribed to: "+topicNumber;

      // if a node identifier is not assigned yet to this link, remove the topicNumber from
      // subscriptionList and return
      if(nodeIdentifier == 0){
         synchronized(synchObject){
            subscriptionList.remove(new Integer(topicNumber));
         }
         return null;
      }

      // unsubscribe this topicNumber
      transportHandler.dataReceived( removeSubscription(topicNumber), linkId );
      transportHandler.dataReceived( removeSubscription(topicNumber + 1), linkId );
      subscriptionList.remove(new Integer(topicNumber));
      return null;
   }

   private byte[] createSubscription(int subscription) {
      RtpProfile rtpProfile = new RtpProfile(subscription, brokerLocation, nodeIdentifier);
      byte[] profileBytes = rtpProfile.getBytes();
      int profileLength = profileBytes.length;
      ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
      DataOutputStream dout = new DataOutputStream(baOutputStream);

      try {
         dout.writeByte(Protocol.RTP_CLIENT_ADD_PROFILE);
         dout.writeInt(profileLength);
         dout.write(profileBytes);
      } catch (IOException ioe) {
         System.out.println(moduleName + "Error marshalling stream" + ioe);
      }/* end try-catch */

      return baOutputStream.toByteArray();
   }/*end method createSubscription() */

   private byte[] removeSubscription(int subscription) {
      RtpProfile rtpProfile = new RtpProfile(subscription, brokerLocation, nodeIdentifier);
      byte[] profileBytes = rtpProfile.getBytes();
      int profileLength = profileBytes.length;
      ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
      DataOutputStream dout = new DataOutputStream(baOutputStream);

      try {
         dout.writeByte(Protocol.RTP_CLIENT_REMOVE_PROFILE);
         dout.writeInt(profileLength);
         dout.write(profileBytes);
      } catch (IOException ioe) {
         System.out.println(moduleName + "Error marshalling stream" + ioe);
      }/* end try-catch */

      return baOutputStream.toByteArray();
   }/*end method createSubscription() */

}
